package com.hrsaas.ext.spider.core;

import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import com.hrsaas.ext.spider.core.url.URL;
import com.hrsaas.ext.spider.core.url.URLParser;
import com.hrsaas.ext.spider.err.RedirectException;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.CompositeByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.HttpContent;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;

/**
 * <p>
 * 功能：response编码解码
 * <p>
 * 
 * @author wangxiaoliang@kanzhun.com
 *
 *         2018年7月31日
 */
@Sharable
final class MessageHandlerImpl extends ChannelInboundHandlerAdapter {

	private static final int DEFAULT_MAX_REDIRECTS = 15;
	private final boolean followRedirects;
	private final HttpClient client;
	private final int maxRedirects;

	MessageHandlerImpl(boolean followRedirects, HttpClient client, int maxRedirects) {
		this.followRedirects = followRedirects;
		this.client = client;
		this.maxRedirects = maxRedirects == -1 ? DEFAULT_MAX_REDIRECTS : maxRedirects;
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		RequestInfo info = ctx.channel().attr(HttpClient.KEY).get();
		if (!info.cancelled.get()) {
			// Premature close, which is a legitimate way of ending a request
			// with no chunks and no Content-Length header according to RFC
			try {
				sendFullResponse(ctx);
			} finally {
				info.handle.event(new State.Closed());
			}
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		RequestInfo info = ctx.channel().attr(HttpClient.KEY).get();
		info.handle.event(new State.Error(cause));
	}

	private boolean checkCancelled(ChannelHandlerContext ctx) {
		RequestInfo info = ctx.channel().attr(HttpClient.KEY).get();
		boolean result = info != null && info.cancelled.get();
		if (result) {
			Channel ch = ctx.channel();
			if (ch.isOpen()) {
				ch.close();
			}
		}
		return result;
	}

	static class ResponseState {

		private final CompositeByteBuf aggregateContent;
		volatile HttpResponse resp;
		volatile boolean fullResponseSent;
		private int readableBytes;
		private final boolean dontAggregate;

		ResponseState(ChannelHandlerContext ctx, boolean dontAggregate) {
			aggregateContent = ctx.alloc().compositeBuffer();
			this.dontAggregate = dontAggregate;
		}

		int readableBytes() {
			return readableBytes;
		}

		void append(ByteBuf buf) {
			readableBytes += buf.readableBytes();
			if (!dontAggregate) {
				int ix = aggregateContent.writerIndex();
				int added = buf.readableBytes();
				aggregateContent.addComponent(buf);
				aggregateContent.writerIndex(ix + added);
			}
		}

		boolean hasResponse() {
			return resp != null;
		}
	}

	public static final AttributeKey<ResponseState> RS = AttributeKey.<ResponseState>valueOf("state");

	private ResponseState state(ChannelHandlerContext ctx, RequestInfo info) {
		Attribute<ResponseState> st = ctx.channel().attr(RS);
		ResponseState rs = st.get();
		if (rs == null) {
			rs = new ResponseState(ctx, info.dontAggregate);
			st.set(rs);
		}
		return rs;
	}

	/**
	 * 是否需要进行重定向
	 * @param info
	 * @param msg
	 * @return
	 * @throws UnsupportedEncodingException
	 */
	private String isRedirect(RequestInfo info, HttpResponse msg) throws UnsupportedEncodingException {
		HttpResponseStatus status = msg.status();
		switch (status.code()) {
		case 300:
		case 301:
		case 302:
		case 303:
		case 305:
		case 307:
			String hdr = msg.headers().get(HttpHeaderNames.LOCATION);
			hdr = URLDecoder.decode(hdr);
			if (hdr != null) {
				if (hdr.toLowerCase().startsWith("https://") ||
						hdr.toLowerCase().startsWith("http://")) {
					return hdr;
				} else {
					URL orig = info.url;
					String pth = orig.getPath() == null ? "/"
							: URLDecoder.decode(orig.getPath().toString(), "UTF-8");
					if (hdr.startsWith("/")) {
						pth = hdr;
					} else if (pth.endsWith("/")) {
						pth += hdr;
					} else {
						pth += "/" + hdr;
					}
					StringBuilder sb = new StringBuilder(orig.getProtocol().getName());
					sb.append("://").append(orig.getHost());
					sb.append(":").append(orig.getProtocol().getDefaultPort());
					if (pth.charAt(0) != '/') {
						sb.append('/');
					}
					sb.append(pth);
					return sb.toString();
				}
			}
		default:
			return null;
		}
	}

	@Override
	public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
		final RequestInfo info = ctx.channel().attr(HttpClient.KEY).get();
		if (checkCancelled(ctx)) {
			return;
		}
		final ResponseState state = state(ctx, info);
		if (msg instanceof HttpResponse) {
			state.resp = (HttpResponse) msg;
			boolean redirectLoop = false;
			String redirUrl = null;
			if (followRedirects) {
				redirUrl = isRedirect(info, state.resp);
				if (redirUrl != null) {
					HttpMethod meth = state.resp.status().code() == 302 ? HttpMethod.GET
							: HttpMethod.valueOf(info.req.method().name());
					// Shut off events from the old request
					AtomicBoolean ab = new AtomicBoolean(true);
					RequestInfo b = new RequestInfo(info.url, info.req, ab, new ResponseFuture(ab),
							null, info.timeout, info.timer, info.dontAggregate,
							info.chunkedBody);
					ctx.channel().attr(HttpClient.KEY).set(b);
					if (info.redirectCount.getAndIncrement() < maxRedirects) {
						URL url;
						if (redirUrl.contains("://")) {
							url = URLParser.getURL(redirUrl);
						} else {
							if (redirUrl.length() > 0 && redirUrl.charAt(0) == '/') {
								url = URLParser.getURL(redirUrl);
							} else if (redirUrl.length() > 0 && redirUrl.charAt(0) != '/') {
								url = URLParser.getURL(info.url.toString() + redirUrl);
							} else {
								url = URLParser.getURL(redirUrl);
							}
						}
						info.handle.event(new State.Redirect(url));
						info.handle.cancelled = new AtomicBoolean();
						// set request url to request buider
						info.r.setRedirectUrl(redirUrl);
						client.redirect(meth, url, info, state(ctx, info), ctx);
						return;
					} else {
						redirectLoop = true;
					}
				}
			}
			info.handle.event(new State.HeadersReceived(state.resp));
			if (redirectLoop) {
				RedirectException redirException = new RedirectException(
						RedirectException.Kind.REDIRECT_LOOP, redirUrl);
				info.handle.event(new State.Error(redirException));
				info.handle.cancelled.lazySet(true);
			}
			// no response content : content length is 0 ,will be next 
			String conlen = state.resp.headers().get(HttpHeaderNames.CONTENT_LENGTH);
			if (conlen != null && HttpResponseStatus.OK.equals(state.resp.status())
					&& Integer.valueOf(conlen) == 0) {
				sendNoResponse(ctx);
			}
		} else if (msg instanceof HttpContent) {
			if (msg instanceof LastHttpContent && state.resp != null
					&& (HttpResponseStatus.CONTINUE.equals(state.resp.status()))) {
				info.handle.event(new State.AwaitingResponse());
				return;
			}
			HttpContent c = (HttpContent) msg;
			info.handle.event(new State.ContentReceived(c));
			c.content().resetReaderIndex();
			if (c.content().readableBytes() > 0) {
				state.append(c.content());
			}
			state.aggregateContent.resetReaderIndex();
			boolean last = c instanceof LastHttpContent;
			if (!last && state.resp.headers().get(HttpHeaderNames.CONTENT_LENGTH) != null) {
				long len = Long.valueOf(state.resp.headers().get(HttpHeaderNames.CONTENT_LENGTH));
				last = state.readableBytes() >= len;
			}
			if (last) {
				c.content().resetReaderIndex();
				sendFullResponse(ctx);
			}
		} else {
			info.handle.event(new State.Error(new IllegalStateException("Unknown object decoded: " + msg)));
		}
	}

	void sendFullResponse(ChannelHandlerContext ctx) {
		RequestInfo info = ctx.channel().attr(HttpClient.KEY).get();
		if (info != null && info.dontAggregate) {
			return;
		}
		ResponseState state = state(ctx, info);
		Class<? extends State<?>> type = State.Finished.class;
		state.aggregateContent.resetReaderIndex();
		if (info != null) {
			info.cancelTimer();
		}
		if ((info.r != null || info.handle.has(type)) && !state.fullResponseSent
				&& state.aggregateContent.readableBytes() > 0) {
			state.fullResponseSent = true;
			info.handle.event(new State.FullContentReceived(state.aggregateContent));
			DefaultFullHttpResponse full = new DefaultFullHttpResponse(state.resp.protocolVersion(),
					state.resp.status(), state.aggregateContent);
			for (Map.Entry<String, String> e : state.resp.headers().entries()) {
				full.headers().add(e.getKey(), e.getValue());
			}
			state.aggregateContent.resetReaderIndex();

			if (info.r != null) {
				info.r.internalReceive(state.resp.status(), state.resp.headers(),
						state.aggregateContent);
			}
			state.aggregateContent.resetReaderIndex();
			info.handle.event(new State.Finished(full));
			state.aggregateContent.resetReaderIndex();
			info.handle.trigger();
		}
	}
	
	/**
	 * no response content but has less response header and cookie
	 * @param ctx
	 */
	void sendNoResponse(ChannelHandlerContext ctx) {
		RequestInfo info = ctx.channel().attr(HttpClient.KEY).get();
		ResponseState state = state(ctx, info);
		if (info != null) {
			info.cancelTimer();
		}
		state.fullResponseSent = true;
		DefaultFullHttpResponse full = new DefaultFullHttpResponse(state.resp.protocolVersion(),
				state.resp.status(), state.aggregateContent);
		for (Map.Entry<String, String> e : state.resp.headers().entries()) {
			full.headers().add(e.getKey(), e.getValue());
		}
		info.r.internalReceive(state.resp.status(), state.resp.headers(),null);
		info.handle.event(new State.Finished(full));
		info.handle.trigger();
	}

}
